package com.xzj.admin.resource.analysis.handle.analysis.pg;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson2.JSON;
import com.xzj.admin.analysis.base.common.cache.WaitDelCache;
import com.xzj.admin.analysis.base.constants.CommonEnums;
import com.xzj.admin.analysis.base.domain.CommonFlowInfo;
import com.xzj.admin.analysis.base.domain.ResMessageManage;
import com.xzj.admin.analysis.base.util.AnalysisCommonUtil;
import com.xzj.admin.resource.analysis.handle.async.AsyncToPgHandle;
import com.xzj.admin.resource.analysis.handle.redis.CustomRedisHandle;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;

/**
 * @program: xzj-admin
 * @ClassName: MsgBatchToPg
 * @description: 消息入pg库
 * @author: XiaZunJun
 * @create: 2024-11-20 15:20
 */
@Slf4j
@Component
public class MsgBatchToPg {

    private static final Map<String, Map<String, List<ResMessageManage>>> MESSAGE_MAP = new ConcurrentHashMap<>();
    private static final Lock JUDGE_ROW_LOCK = new ReentrantLock();
    @Resource
    private AsyncToPgHandle asyncToPgHandle;

    @Resource
    private CustomRedisHandle<String> customRedisHandle;

    /**
     * 判断消息新增修改
     *
     * @param row 单个入库对象
     * @author xiazunjun
     * @date 2024/3/26 9:02
     */
    public void judgeRowTag(CommonFlowInfo info, ResMessageManage row) {
        // 拼接一级key
        Map<String, String> md5Map = WaitDelCache.getMd5Map(row.getMessageSource(), row.getRmPoolId());
        String cacheKey = md5Map.get(CommonEnums.CACHE_KEY.PREFIX.getValue());
        try {
            JUDGE_ROW_LOCK.lock();
            String messageKey = AnalysisCommonUtil.commonBuildStr(info.getThreadName(), cacheKey);
            if (!MESSAGE_MAP.containsKey(messageKey)) {
                Map<String, List<ResMessageManage>> sourceMessageMap = new HashMap<>();
                sourceMessageMap.put(CommonEnums.MESSAGE_TAG.edit.name(), new ArrayList<>());
                sourceMessageMap.put(CommonEnums.MESSAGE_TAG.add.name(), new ArrayList<>());
                MESSAGE_MAP.put(messageKey, sourceMessageMap);
            }
            if (md5Map.containsKey(row.getMessageId())) {
                if (!md5Map.get(row.getMessageId()).equals(row.getMessageJsonMd5())) {
                    row.setMessageTag(CommonEnums.MESSAGE_TAG.edit.name());
                    putMessage(cacheKey,messageKey, CommonEnums.MESSAGE_TAG.edit, row);
                }
                md5Map.remove(row.getMessageId());
            } else {
                Object cacheMapValue = customRedisHandle.getCacheMapValue(cacheKey, row.getMessageId());
                if (ObjectUtil.isEmpty(cacheMapValue)) {
                    row.setMessageTag(CommonEnums.MESSAGE_TAG.add.name());
                    putMessage(cacheKey,messageKey, CommonEnums.MESSAGE_TAG.add, row);
                }
            }
        } catch (Exception e) {
            log.error("exception in judgeRowTag,{} ", JSON.toJSONString(row), e);
        } finally {
            JUDGE_ROW_LOCK.unlock();
        }
    }
    private void putMessage(String cacheKey,String messageKey, CommonEnums.MESSAGE_TAG messageTag, ResMessageManage row) {
        Map<String, List<ResMessageManage>> messageMap = MESSAGE_MAP.get(messageKey);
        if (null != messageMap) {
            List<ResMessageManage> list = messageMap.get(messageTag.name());
            if (null == list) {
                list = new ArrayList<>();
            }
            list.add(row);
            messageMap.put(messageTag.name(), list);
        } else {
            messageMap = new HashMap<>();
            List<ResMessageManage> list = new ArrayList<>();
            list.add(row);
            messageMap.put(messageTag.name(), list);
        }
        customRedisHandle.setCacheMapValue(cacheKey, row.getMessageId(), row.getMessageJsonMd5());
    }

    public void batchToPg(CommonFlowInfo info, Map<String, String> md5Map) {
        String cacheKey = md5Map.get(CommonEnums.CACHE_KEY.PREFIX.getValue());
        if (StrUtil.isNotEmpty(cacheKey)) {
            cacheKey = AnalysisCommonUtil.commonBuildStr(info.getThreadName(), cacheKey);
            Map<String, List<ResMessageManage>> msgListMap = MESSAGE_MAP.remove(cacheKey);
            if (null == msgListMap) {
                return;
            }
            for (String type : msgListMap.keySet()) {
                List<ResMessageManage> batchList = msgListMap.get(type);
                // 在传递到MyBatis前过滤掉某字段为空的对象
                batchList = batchList.stream()
                        .filter(obj -> obj.getMessageId() != null)
                        .collect(Collectors.toList());
                List<List<ResMessageManage>> listList = ListUtil.split(batchList, 100);
                for (List<ResMessageManage> pgList : listList) {
                    asyncToPgHandle.batchAddOrUpdate(pgList, type);
                }
            }
        }

    }
}
